package com.chinamobile.iot.handler;

import com.alibaba.fastjson.JSON;
import com.chinamobile.iot.config.BrokerConfiguration;
import com.chinamobile.iot.config.MqttConfiguration;
import com.chinamobile.iot.handler.model.RedisClientSessionSub;
import com.chinamobile.iot.handler.model.Tree3Client;
import com.chinamobile.iot.mqtt.interception.MessageHandler;
import com.chinamobile.iot.mqtt.spi.impl.ProtocolProcessor;
import com.chinamobile.iot.mqtt.spi.impl.SubscriptionTree;
import com.chinamobile.iot.mqtt.spi.model.Qos1Message;
import com.chinamobile.iot.mqtt.spi.model.StoredMessage;
import com.chinamobile.iot.mqtt.spi.model.Tree1Client;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.LinkedList;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import static com.chinamobile.iot.handler.redis.RedisConst.REDIS_3_KEY_PREFIX;
import static com.chinamobile.iot.handler.redis.RedisConst.REDIS_4_KEY_PREFIX;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE;

/**
 * Created by szl on 2017/7/31.
 * <p>
 * when the broker is run in standalone mode, it will use this to processor clean session hand offline message
 */
@Component("extendMessageHandler")
@ConditionalOnExpression("${mqtt.distribute} == false")
public class StandaloneMessageHandler implements MessageHandler {

    private static Logger logger = LoggerFactory.getLogger(StandaloneMessageHandler.class);

    @Autowired
    private BrokerConfiguration brokerConfiguration;
    /**
     * 本地使用的订阅树
     */
    private SubscriptionTree<Tree1Client> tree1;
    /**
     * the subscription tree of global, used for CleanSession = false
     */
    private SubscriptionTree<Tree3Client> tree3;

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * used by addAllClientToZk, hasClientConnected
     */
    @Autowired
    private ProtocolProcessor protocolProcessor;
    @Autowired
    private MqttConfiguration mqttConfiguration;


    @Override
    public boolean init() {
        tree1 = protocolProcessor.getTree1();
        tree3 = new SubscriptionTree<>();
        return true;
    }

    @Override
    public void destroy() {

    }

    @Override
    public boolean onConnecting(String clientId, String mqttId) {
        return true;
    }

    @Override
    public void onConnected(String clientId, String mqttId, boolean cleanSession) {
        // 读取以前是否有保存的QoS订阅
        if (mqttConfiguration.isCleanSession() && mqttConfiguration.isSubscribe()) {
            SetOperations<String, String> setOperations = redisTemplate.opsForSet();
            String redis3Key = REDIS_3_KEY_PREFIX + clientId;
            Set<String> result = setOperations.members(redis3Key);

            if (result != null && !result.isEmpty()) {
                if (cleanSession) {
                    result.forEach((sub) -> {
                        RedisClientSessionSub redisClientSessionSub = JSON.parseObject(sub, RedisClientSessionSub.class);

                        int qos = redisClientSessionSub.getQ();
                        String filter = redisClientSessionSub.getF();
                        String broker = redisClientSessionSub.getB();
                        // 加入本地订阅树
                        Tree1Client client1 = new Tree1Client(clientId, true);
                        tree1.add(filter, client1, qos);
                        Tree3Client client3 = new Tree3Client(clientId, broker);
                        tree3.removeSubscribe(client3);
                    });
                    redisTemplate.delete(redis3Key);
                } else {
                    LinkedList<String> newSub = new LinkedList<>();
                    result.forEach((sub) -> {
                        RedisClientSessionSub redisClientSessionSub = JSON.parseObject(sub, RedisClientSessionSub.class);
                        int qos = redisClientSessionSub.getQ();
                        String filter = redisClientSessionSub.getF();
                        String broker = redisClientSessionSub.getB();
                        Tree1Client client1 = new Tree1Client(clientId, false);
                        tree1.add(filter, client1, qos);
                        Tree3Client client3 = new Tree3Client(clientId, broker);
                        tree3.removeSubscribe(client3);
                        redisClientSessionSub.setB(brokerConfiguration.getName());
                        newSub.add(JSON.toJSONString(redisClientSessionSub));
                    });
                    // 在redis3先删除记录，再新增，达到修改其broker信息的效果
                    redisTemplate.delete(redis3Key);
                    newSub.forEach((s) -> setOperations.add(redis3Key, s));
                }
            }
        }
    }

    @Override
    public void onClientDisconnected(String clientId, String mqttId, boolean cleanSession) {
        if (mqttConfiguration.isSubscribe()) {
            processDisconnect(clientId, mqttId, cleanSession);
        }
    }

    @Override
    public void onTopicPublished(String clientId, String mqttId, StoredMessage msg, String filter, int subCount) {
        if (mqttConfiguration.isCleanSession() && mqttConfiguration.isSubscribe()) {
            String topic = msg.getTopic();

            Set<SubscriptionTree.ClientInfo<Tree3Client>> clientInfos;
            // 检查tree3
            clientInfos = tree3.matchFull(topic);
            if (clientInfos != null && !clientInfos.isEmpty()) {
                clientInfos.forEach((c) -> {
                    Tree3Client client = c.getClient();
                    int qos = msg.getQos().value() < c.getQos() ? msg.getQos().value() : c.getQos();
                    if (qos == AT_LEAST_ONCE.value()) {
                        // 包数据保存在redis中
                        saveMsgForClient(client.getC(), msg, msg.getMessageId());
                    }
                });
            }
        }
    }

    @Override
    public void onTopicSubscribed(String clientId, String mqttId, String topicFilter, int qos, boolean cleanSession, int oldQos) {
        if (mqttConfiguration.isCleanSession() && mqttConfiguration.isSubscribe()) {
            SetOperations<String, String> setOperations = redisTemplate.opsForSet();
            if (!cleanSession) {
                // 在Redis3中添加记录
                String clientKey = REDIS_3_KEY_PREFIX + clientId;
                RedisClientSessionSub clientSub = new RedisClientSessionSub();
                clientSub.setB(brokerConfiguration.getName());
                clientSub.setF(topicFilter);

                if (oldQos != 3 && oldQos != qos){
                    clientSub.setQ(oldQos);
                    setOperations.remove(clientKey, JSON.toJSONString(clientSub));
                }

                clientSub.setQ(qos);
                setOperations.add(clientKey, JSON.toJSONString(clientSub));
            }
        }
    }

    @Override
    public void onTopicUnsubscribed(String topic, int qos, String clientId, boolean cleanSession) {
        if (mqttConfiguration.isCleanSession() && mqttConfiguration.isSubscribe()) {
            SetOperations<String, String> setOperations = redisTemplate.opsForSet();
            if (!cleanSession) {
                // 从Redis3中删除订阅信息
                RedisClientSessionSub clientSessionSub = new RedisClientSessionSub();
                clientSessionSub.setB(brokerConfiguration.getName());
                clientSessionSub.setF(topic);
                clientSessionSub.setQ(qos);
                String key = REDIS_3_KEY_PREFIX + clientId;
                setOperations.remove(key, JSON.toJSONString(clientSessionSub));
            }
        }
    }

    @Override
    public void onMessageAcknowledged(String clientId, String mqttId, int messageId) {

    }

    @Override
    public void onRouteMessageToClient(String clientId, String mqttId, String topic) {

    }

    private void processDisconnect(String clientId, String mqttId, boolean cleanSession) {
        if (mqttConfiguration.isCleanSession() && !cleanSession) {
            // 读取redis3中该客户端的所有全局订阅
            SetOperations<String, String> setOperations = redisTemplate.opsForSet();
            String redis3Key = REDIS_3_KEY_PREFIX + clientId;
            Set<String> subs = setOperations.members(redis3Key);
            LinkedList<String> clientSessionSubs = new LinkedList<>();
            if (subs != null && subs.size() > 0) {
                subs.forEach(sub -> clientSessionSubs.add(sub));
            }
            clientSessionSubs.forEach((s) -> {
                RedisClientSessionSub sub = JSON.parseObject(s, RedisClientSessionSub.class);
                // 修改redis3中所有记录，设置其broker为null
                setOperations.remove(redis3Key, JSON.toJSONString(sub));
                sub.setB(null);
                setOperations.add(redis3Key, JSON.toJSONString(sub));
                Tree3Client client = new Tree3Client(clientId, sub.getB());
                tree3.add(sub.getF(), client, sub.getQ());
            });
        }
        Tree1Client client1 = new Tree1Client(clientId, cleanSession);
        tree1.removeSubscribe(client1);
    }

    private void saveMsgForClient(String clientId, StoredMessage message, Integer messageId) {
        Qos1Message qos1Message = new Qos1Message();
        qos1Message.setRound(1);
        qos1Message.setToClientId(clientId);
        qos1Message.setMsgId(messageId);
        qos1Message.setData(message);

        String key = REDIS_4_KEY_PREFIX + clientId;

        ListOperations<String, String> listOperations = redisTemplate.opsForList();
        long size = listOperations.size(key);
        if (size == 0) {
            listOperations.rightPush(key, JSON.toJSONString(qos1Message));
            redisTemplate.expire(key, 3, TimeUnit.DAYS);
        } else if (size < 500) {
            listOperations.rightPush(key, JSON.toJSONString(qos1Message));
        }
    }
}
